package org.sg20.pagerank.pagerankiter;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.sg20.utils.Edge;
import org.sg20.utils.ParseText;

import java.io.IOException;

public class PageRankIterMapper extends Mapper<LongWritable, Text, Text, Text>{

    Text out_key = new Text();

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException{
        super.setup(context);
    }


    /*
    由GraphBuilder的输出得到格式为 唐僧 1.0|悟空,八戒
    本阶段的输出唐僧 |1.0|悟空*0.5,八戒*0.5  or   唐僧 pr
     */
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
        String[] thisInput = value.toString().split("\t");
        // pr|related_person
        String line = thisInput[1];
        //将pr和链接到的人物划分开
        String[] tuple = line.split("\\|");
        //System.out.println(tuple[0] + " " + thisInput[0] + " " + thisInput[1]);
        Double pr = Double.parseDouble(tuple[0]);
        out_key.set(thisInput[0]);
        if(tuple.length > 1) {
            //split后是悟空*0.5
            String[] linkPeople = tuple[1].split(",");
            for (String linkPerson : linkPeople) {
                String[] personWeight = linkPerson.split("\\*");
                String person = personWeight[0];
                double weight = Double.parseDouble(personWeight[1]);
                String prVal = thisInput[0] + "\t" + String.valueOf(pr * weight);
                //输出传递给该网页的PR值
                context.write(new Text(person), new Text(prVal));
            }
            //区分两种输出(传出link_list)
            context.write(out_key, new Text("|" + tuple[1]));
        }
    }
}
